package com.shujia.batch.car

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._
import org.apache.spark.sql.{DataFrame, SparkSession}

object DwsAppVioZqzf {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .appName("DwsAppVioZqzf")
      .enableHiveSupport() //开启hive元数据支持
      .getOrCreate()
    import spark.implicits._
    import org.apache.spark.sql.functions._

    /**
      * 1、读取现场违法表
      *
      */

    val baseVioForce: DataFrame = spark
      .table("dwd.dwd_base_vio_force")
      .where($"ds" === "20220627")


    /**
      * 2、非现场违法表
      *
      */
    val vioSurveil: DataFrame = spark
      .table("dwd.dwd_base_vio_surveil")
      .where($"ds" === "20220627")


    /**
      * 2、取出现场违法表中所有的违法行为
      *
      */

    val unionDF: DataFrame = baseVioForce
      .select(
        date_format($"wfsj", "yyyy-MM-dd") as "wfsj", //违法时间
        explode(array($"wfxw1", $"wfxw2", $"wfxw3", $"wfxw4", $"wfxw5")) as "wfxw" //取出一次违法中多个违法行为

      )
      .union(vioSurveil.select(date_format($"wfsj", "yyyy-MM-dd") as "wfsj", $"wfxw")) //合并非现场违法表
      .where($"wfxw".isNotNull) //过滤违法行为为空
      .where($"wfsj".isNotNull) //过滤违法时间为空的数据


    var indexDF: DataFrame = unionDF
      //按照日期分组
      .groupBy($"wfsj")
      .agg(
        //当日违法数
        count("wfxw") as "dr_zd_wfs",
        //当日醉酒驾驶违法数
        sum(when($"wfxw".isin("20011", "20021", "1702", "1703", "6022", "5049", "2010", "20102", "20010", "20020", "17021", "17031", "6032", "6033", "10043", "A01"), 1).otherwise(0)) as "dr_zjjs_wfs",
        //当饮酒驾驶违法数
        sum(when($"wfxw".isin("20061", "20071", "1711", "1712", "1713", "6022", "A17", "20060", "20070", "17122", "17121", "1605", "1604", "17111", "6034", "6035", "60351", "S03", "10043"), 1).otherwise(0)) as "dr_yjjs_wfs",
        //当日遮挡号牌
        sum(when($"wfxw".isin("20211", "1614", "1329", "20210", "1718", "17181", "1331", "1616", "17201", "1720"), 1).otherwise(0)) as "dr_zdhp_wfs",
        //当日伪造变造违法数
        sum(when($"wfxw".isin("75011", "5702", "5703", "5054", "5706", "5002", "5003", "5006", "5007", "5010", "5011", "5601", "5701", "5009", "57011", "57012", "57031", "57032", "50101", "50102", "50041", "50042", "50051", "50052", "50012", "50011", "50062", "50061", "50072", "50071", "50112", "50111", "5001", "50031", "50021", "5004", "5005", "5008", "5012", "57022", "57021", "50022", "50032", "60021"), 1).otherwise(0)) as "dr_wzbz_wfs",
        //当日套牌车违法数
        sum(when($"wfxw".isin("57011", "57012", "5054", "5706", "5601", "5701", "5002", "50021", "5008", "50022"), 1).otherwise(0)) as "dr_tpjp_wfs",
        //当日未戴头盔违法数
        sum(when($"wfxw".isin("3020", "30201", "1207", "20301", "12071", "74021"), 1).otherwise(0)) as "dr_wdtk_wfs",
        //当日驾驶拼装报废机动车违法数
        sum(when($"wfxw".isin("6021", "5051", "10022", "1002", "20671"), 1).otherwise(0)) as "dr_jspz_wfs",
        //当日无有效驾驶资格违法数
        sum(when($"wfxw".isin("6026", "20111", "10072", "10062", "5703", "1010", "5006", "20032", "10052", "16103", "10064", "1015", "10151", "1610", "1709", "1009", "10091", "10061", "10141", "1014", "57031", "57032", "1005", "1006", "10104", "17091", "10051", "16102", "16104", "16101", "50062", "50061", "1704", "10821", "1082", "10101", "10103", "20110", "5012", "17094", "17093", "10054", "10053", "10063", "10102", "17092", "60063", "60061", "60062", "60021"), 1).otherwise(0)) as "dr_wxjs_wfs",
        //当日超员违法数
        sum(when($"wfxw".isin("1710", "20081", "20082", "1601", "17101", "17103", "1202", "1621", "16211", "17102", "17104", "20231", "20412", "6038", "1238", "1348", "13481", "71171", "1241", "1341", "1714", "1623", "1626", "16261", "1627", "17161", "6017", "60171", "1716", "16271"), 1).otherwise(0)) as "dr_cy_wfs",
        //当日超载违法数
        sum(when($"wfxw".isin("13542", "1342", "1201", "13532", "1637", "71181", "13534", "13544", "13543", "13541", "16391", "16393", "16392", "16394", "12011", "1602", "13421", "13422", "12012", "13533", "13531", "1353", "20241", "1639", "16371", "16373", "16374", "16372", "1354", "1346", "1239", "1608"), 1).otherwise(0)) as "dr_cz_wfs"
      )


    val indexs = List("zd", "zjjs", "yjjs", "zdhp", "wzbz", "tpjp", "wdtk", "jspz", "wxjs", "cy", "cz")


    for (index <- indexs) {
      //循环计算指标
      indexDF = genIndex(indexDF, spark, index)
    }

    //整理数据
    val resultDF: DataFrame = indexDF.select(
      $"wfsj",
      $"dr_zd_wfs",
      $"jn_zd_wfs",
      $"zd_tb",
      $"zd_tbbj",
      $"dr_zjjs_wfs",
      $"jn_zjjs_wfs",
      $"zjjs_tb",
      $"zjjs_tbbj",
      $"dr_yjjs_wfs",
      $"jn_yjjs_wfs",
      $"yjjs_tb",
      $"yjjs_tbbj",
      $"dr_zdhp_wfs",
      $"jn_zdhp_wfs",
      $"zdhp_tb",
      $"zdhp_tbbj",
      $"dr_wzbz_wfs",
      $"jn_wzbz_wfs",
      $"wzbz_tb",
      $"wzbz_tbbj",
      $"dr_tpjp_wfs",
      $"jn_tpjp_wfs",
      $"tpjp_tb",
      $"tpjp_tbbj",
      $"dr_wdtk_wfs",
      $"jn_wdtk_wfs",
      $"wdtk_tb",
      $"wdtk_tbbj",
      $"dr_jspz_wfs",
      $"jn_jspz_wfs",
      $"jspz_tb",
      $"jspz_tbbj",
      $"dr_wxjs_wfs",
      $"jn_wxjs_wfs",
      $"wxjs_tb",
      $"wxjs_tbbj",
      $"dr_cy_wfs",
      $"jn_cy_wfs",
      $"cy_tb",
      $"cy_tbbj",
      $"dr_cz_wfs",
      $"jn_cz_wfs",
      $"cz_tb",
      $"cz_tbbj"
    )

    //保存数据到表中
    resultDF.createOrReplaceTempView("tmp")
    spark.sql(
      """
        |insert overwrite table dws.dws_app_vio_zqzf partition (ds='20220627')
        |select * from tmp
      """.stripMargin)

  }

  def genIndex(df: DataFrame, spark: SparkSession, index: String): DataFrame = {
    import spark.implicits._
    df.withColumn(s"jn_${index}_wfs", sum($"dr_${index}_wfs") over Window.partitionBy(year($"wfsj"))) //今年违法数
      .withColumn(s"qn_dr_${index}_wfs", lag($"dr_${index}_wfs", 1, 1) over Window.partitionBy(date_format($"wfsj", "MM-dd")).orderBy(year($"wfsj")))
      .withColumn(s"${index}_tb", coalesce(round(($"dr_${index}_wfs" - $"qn_dr_${index}_wfs") / $"qn_dr_${index}_wfs" * 100, 2), expr("0.0")))
      .withColumn(s"${index}_tbbj", when($"dr_${index}_wfs" > $"qn_dr_${index}_wfs", "上升").otherwise("下降"))

  }

}
